package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.service.CKDBService;
import com.atguigu.upp.service.MySQLDBService;
import com.atguigu.upp.utils.TagValueTypeConstant;
import com.atguigu.upp.utils.UPPUtil;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/12/3
 */
public class WriteBitmapApp
{
    public static void main(String[] args) throws IOException {

        //①接收参数
        String taskId = args[0];
        String doDate = args[1];

     /*   String taskId = "3";
        String doDate = "2020-06-14";*/

        //②获取ckDBService
        SqlSessionFactory sqlSessionFactory = UPPUtil.createSqlSessionFactory("mysql_config.xml");
        MySQLDBService mySQLDBService = new MySQLDBService(sqlSessionFactory.openSession());
        SqlSessionFactory ckSSF = UPPUtil.createSqlSessionFactory("ck_config.xml");
        CKDBService ckdbService = new CKDBService(ckSSF.openSession());

        //获取今天宽表中计算的标签
        List<TagInfo> tags = mySQLDBService.getTagInfoExecute();

        //将计算的标签，找tagValueType进行分类，分为四类。
        //tag_population_attributes_nature_gender,tag_population_attributes_nature_period
        List<TagInfo> stringTags =new ArrayList<>();
        //tag_consumption_behavior_order_7damount
        List<TagInfo> decimalTags =new ArrayList<>();
        List<TagInfo> longTags =new ArrayList<>();
        List<TagInfo> dateTags =new ArrayList<>();

        for (TagInfo tag : tags) {

            switch (tag.getTagValueType()){
                case TagValueTypeConstant.TAG_VALUE_TYPE_LONG: longTags.add(tag); break;
                case TagValueTypeConstant.TAG_VALUE_TYPE_DECIMAL:  decimalTags.add(tag); break;
                case TagValueTypeConstant.TAG_VALUE_TYPE_STRING:  stringTags.add(tag); break;
                case TagValueTypeConstant.TAG_VALUE_TYPE_DATE:  dateTags.add(tag); break;
            }
        }

        //每一类的标签，计算bitmap，写入到目标表
        writeBitmap(stringTags,ckdbService,"user_tag_value_string",doDate);
        writeBitmap(decimalTags,ckdbService,"user_tag_value_decimal",doDate);
        writeBitmap(longTags,ckdbService,"user_tag_value_long",doDate);
        writeBitmap(dateTags,ckdbService,"user_tag_value_date",doDate);

        //不用，也要准备。因为所有的程序，都是通过Spark-submit提交要yarn。
        // spark-submit只能提交sparkapp，要求程序必须有SparkContext
        SparkSession sparkSession = UPPUtil.createSparkSession("WriteBitmapApp");

    }

    private static void writeBitmap(List<TagInfo> tags,CKDBService ckdbService,String table,String doDate){

        if (tags.size() > 0){

            //先之前，先删除
            ckdbService.cleanHistoryBitmap(table,doDate);

            //确认sourceTable(今天的宽表名)
            String sourceTable = UPPUtil.getProperty("upwideprefix") + doDate.replace("-","_");

            /*
                确认tagSql

                ('tag_consumption_behavior_order_7damount', tag_consumption_behavior_order_7damount)
             */
            String tagSql = tags.stream()
                                 .map(t -> String.format(" ('%s', %s ) ", t.getTagCode().toLowerCase(), t.getTagCode().toLowerCase()))
                                 .collect(Collectors.joining(","));

            //写入
            ckdbService.insertBitmap(table,doDate,tagSql,sourceTable);

        }

    }
}
